package com.loan.common;

import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.HashMap;
import java.util.Map;

/**
 * @author denny.gong
 * @version 1.0
 * @date 2021/5/17 11:07
 **/
public class KafkaProducerUtils {


    public void sendMessage(String servers,String topicName,Object data){
        Map<String, Object> configs = new HashMap<String, Object>();
        configs.put("bootstrap.servers",servers);
        //acks=all表示所有同步节点都返回确认，可靠性最高
        configs.put("acks","all");
        //
        configs.put("retries","1");
        configs.put("delivery.timeout.ms","70000");

        DefaultKafkaProducerFactory defaultKafkaProducerFactory = new DefaultKafkaProducerFactory(configs,
                new StringSerializer(),new StringSerializer());
        KafkaTemplate kafkaTemplate=new KafkaTemplate(defaultKafkaProducerFactory);

        kafkaTemplate.send(topicName,data);
    }

}
